#! /usr/bin/perl -w
package Communicator;

####################################################################
################ Communicator ######################################
############ Runner communicates with database######################
####################################################################

use strict;
use DBUtility;
use Task;
use Message;
use Constants;
use Config::ApacheFormat;

our $VERSION = 1.00;

sub new{
   my ($class) = @_;
   my $self = {
      _id => undef,
      _cid => undef,
      _identity => undef,
		_capability => undef,
      _dbutil => undef
   };
   bless $self;
   return $self;
}

sub init{
   my ($self, $host, $db, $dbuser, $dbpasswd, $identity, $capability) = @_;
	
	#create a connection to the database
   $self->{_dbutil} = new DBUtility();
   return $FAILURE if($self->{_dbutil}->init($host, $db, $dbuser, $dbpasswd)); 

	#store the general information about this runner
   $self->{_identity} = "\'". $identity . "\'";
	$self->{_capability} = $capability;

   return $SUCCESS;   
}

####################################################################################
# Register a communicator at database
# Each communicator represents a runner
# Returns the rid of the runner
####################################################################################
sub register{
   my ($self) = @_;

   #delete old records of this resource.
	$self->{_dbutil}->delete_runner_record($self->{_identity});

	#register runner capability and get a capability id
   my $cid = $self->register_runner_capability($self->{_capability});
   return undef if(!defined($cid)); 
   
	#add runner record into runners table
   my $hash = {capability_id=>$cid, identity_str=>$self->{_identity}};
   if(!$self->{_dbutil}->add_record("runners", $hash)) {
		#get runner id
   	my $hashref = $self->{_dbutil}->query("SELECT id FROM runners WHERE identity_str = $self->{_identity}");
      $self->{_id} = $hashref->{id};

		#update runner id in its corresponding capability record
	  	my $cid_ref = {rid => $self->{_id}};
	  	$self->{_dbutil}->update_record("capabilities", $cid_ref, "id = $cid");

		#returns runner's rid
      return $hashref->{id};
   } else {
      return undef;
   }
}

sub registered {
   my ($self) = @_;
	return defined($self->{_id}) ? 1 : 0;
}

sub unregister{
   my ($self) = @_;
   return $FAILURE unless defined($self->{_dbutil});
	#delete runner record
   return $self->{_dbutil}->execute("DELETE FROM runners WHERE id = $self->{_id}");
	#delete capability record
   return $self->{_dbutil}->execute("DELETE FROM capabilities WHERE rid = $self->{_id}");
}

sub destroy{
   my ($self) = @_;
   return $FAILURE unless defined($self->{_dbutil});

   #delete the registration in db;
   $self->unregister();
   
   #end the connect;
   $self->{_dbutil}->destroy();

   return $SUCCESS;
}

sub update_task_stage {
   my ($self, $tid, $stage) = @_;
   return $FAILURE unless defined($self->{_dbutil});
   return $self->{_dbutil}->execute("UPDATE tasks SET stage = $stage WHERE id = $tid and stage <> $TASK_CANCELLED");
}

sub update_task_stage_and_interrupt {
   my ($self, $tid, $stage, $interrupt) = @_;
   return $FAILURE unless defined($self->{_dbutil});
   return $self->{_dbutil}->execute("UPDATE tasks SET stage = $stage, interrupt = $interrupt WHERE id = $tid");
}

sub update_task_status {
   my ($self, $tid, $status) = @_;
   return $FAILURE unless defined($self->{_dbutil});
   return $self->{_dbutil}->execute("UPDATE tasks SET status = \'$status\' WHERE id = $tid");
}

sub update_task_pid {
   my ($self, $tid, $pid) = @_;
   return $FAILURE unless defined($self->{_dbutil});
   return $self->{_dbutil}->execute("UPDATE tasks SET pid = $pid WHERE id = $tid");
}

sub parse_message {
	my ($self, $msg) = @_;
	my $message = new Message();
	$message->init($msg);
	return $message;
}

sub fetch_task {
   my ($self) = @_; 
   return $FAILURE unless defined($self->{_dbutil});

   #print "SELECT tid, command, logfile FROM task WHERE rid = $self->{_id} AND stage = 0 order by createTime";
   my $hashref = $self->{_dbutil}->query(
		"SELECT id, command, logfile FROM tasks WHERE rid = $self->{_id} AND stage = 0 order by createTime");
   return $hashref;
}

sub fetch_task_interrupts {
   my ($self) = @_; 
   return $FAILURE unless defined($self->{_dbutil});
   my $hashref = $self->{_dbutil}->query_multi(
		"SELECT id, interrupt FROM tasks WHERE rid = $self->{_id} AND stage > $TASK_READY AND interrupt <> 0 AND pid > 0 
		order by createTime", "id");
   return $hashref; 
}

sub fetch_task_executable {
   my ($self) = @_; 
   return $FAILURE unless defined($self->{_dbutil});
   my $task_ret = undef;
   my $cap_ref = $self->{_dbutil}->query("SELECT * FROM capabilities WHERE id = $self->{_cid}");
   #print "SELECT tid, command, logfile FROM task WHERE rid = $self->{_id} AND stage = 0 order by createTime";
   my $hashref = $self->{_dbutil}->query_multi(
		"SELECT id, capabilities, command, logfile FROM tasks 
		WHERE rid = $self->{_id} AND stage = $TASK_SCHEDULED order by createTime", "id");

   foreach my $task_id (sort(keys %$hashref)) {
   	my %cap_hash;
   	my $cap_str = $hashref->{$task_id}->{"capabilities"};
		my @capabilities = split /,/, $cap_str;
		my $unmatched_cap = 0;
	
		foreach my $capability (@capabilities) {
			my @cap_item = split /:/, $capability;
			if($cap_ref->{$cap_item[0]} < $cap_item[1]) {
				$unmatched_cap = 1;
				last;
			}
		}

		if(!$unmatched_cap) {
			$task_ret = $hashref->{$task_id};
			last;
		}
   }
   return $task_ret;
}

sub fetch_order {
   my ($self) = @_; 
   return $FAILURE unless defined($self->{_dbutil});
   my $hashref = $self->{_dbutil}->query_multi("SELECT id, interrupt FROM tasks WHERE rid = $self->{_id}");
   return $hashref;
}

sub register_runner_capability{
	my ($self, $cap_string) = @_;
	my %hash;
	my @capabilities = split /,/, $cap_string;
	foreach my $capability (@capabilities) {
		my @cap_item = split /:/, $capability;
		$hash{$cap_item[0]} = $cap_item[1];
	}

  	if(!$self->{_dbutil}->add_record("capabilities", \%hash)) {
		$self->{_cid} = $self->{_dbutil}->get_last_insert_id("capabilities", "id");
  		return $self->{_cid};
	} else {
		return undef;
	}
}

sub consume_capability {
	my ($self, $cap_string) = @_;
	my %hash;
	my @capabilities = split /,/, $cap_string;
	foreach my $capability (@capabilities) {
		my @cap_item = split /:/, $capability;
		$hash{$cap_item[0]} = $cap_item[1];
	}

	my $cap_ref = $self->{_dbutil}->query("SELECT * FROM capabilities WHERE id = $self->{_cid}");

	delete($cap_ref->{"id"});
	delete($cap_ref->{"rid"});

	foreach my $cap_key (keys %$cap_ref) {
		$cap_ref->{$cap_key}-- if($cap_ref->{$cap_key});
	}

	$self->{_dbutil}->update_record("capabilities", $cap_ref, "id = $self->{_cid}");
}

sub recycle_capability {
	my ($self, $cap_string) = @_;
	my %hash;
	my @capabilities = split /,/, $cap_string;
	foreach my $capability (@capabilities) {
		my @cap_item = split /:/, $capability;
		$hash{$cap_item[0]} = $cap_item[1];
	}

	my $cap_ref = $self->{_dbutil}->query("SELECT * FROM capabilities WHERE id = $self->{_cid}");

	delete($cap_ref->{"id"});
	delete($cap_ref->{"rid"});

	foreach my $cap_key (keys %$cap_ref) {
		$cap_ref->{$cap_key}++;
	}

	$self->{_dbutil}->update_record("capabilities", $cap_ref, "id = $self->{_cid}");
}

sub get_pid_by_tid {
   my ($self, $tid) = @_; 
   return $FAILURE unless defined($self->{_dbutil});
   my $hashref = $self->{_dbutil}->query("SELECT pid FROM tasks WHERE id = $tid");
   return $hashref->{"pid"};
}

